package drivers

import (
	"app/pkg/queue"
	"context"
	"encoding/json"
	"fmt"
	"github.com/go-redis/redis/v8"
	"os"
	"time"
)

type RedisStreamDriver struct {
	client *redis.Client
}

func NewRedisStream(opts *redis.Options) *RedisStreamDriver {
	return &RedisStreamDriver{
		client: redis.NewClient(opts),
	}
}

func (r *RedisStreamDriver) Connect() error {
	return r.client.Ping(context.Background()).Err()
}

func (r *RedisStreamDriver) Close() error {
	return r.client.Close()
}

func (r *RedisStreamDriver) Topic(topic string) error {
	return nil
}

func (r *RedisStreamDriver) Publish(ctx context.Context, topic string, message []byte, delay int) error {
	if delay > 0 {
		return r.client.ZAdd(ctx, r.delayQueueName(topic), &redis.Z{
			Score:  float64(time.Now().Add(time.Duration(delay) * time.Second).Unix()),
			Member: message,
		}).Err()
	}
	// 立即消息发布到Stream
	_, err := r.client.XAdd(ctx, &redis.XAddArgs{
		Stream: r.queueName(topic),
		ID:     "*", //自动生成id
		Values: map[string]interface{}{
			"data": message,
		},
	}).Result()
	return err
}

func (r *RedisStreamDriver) processDelayedMessages(ctx context.Context, topic string) {
	now := time.Now().Unix()
	messages, err := r.client.ZRangeByScore(ctx, r.delayQueueName(topic), &redis.ZRangeBy{
		Min:    "0",
		Max:    fmt.Sprintf("%d", now),
		Offset: 0,
		Count:  10,
	}).Result()
	if err != nil {
		return
	}
	for _, msg := range messages {
		if err := r.Publish(ctx, topic, []byte(msg), 0); err == nil {
			r.client.ZRem(ctx, r.delayQueueName(topic), msg)
		}
	}
}

func (r *RedisStreamDriver) Subscribe(ctx context.Context, topic string, opt queue.Options, callback func(ctx context.Context, channel <-chan queue.QueueMessgae)) error {
	// 创建消费者组
	err := r.client.XGroupCreateMkStream(ctx, topic, opt.ConsumerGroup, "0").Err()
	if err != nil && err.Error() != "BUSYGROUP Consumer Group name already exists" {
		return err
	}
	// 延迟消息处理定时器
	delayTicker := time.NewTicker(1 * time.Second)
	defer delayTicker.Stop()

	msgChan := make(chan queue.QueueMessgae, 1)
	defer close(msgChan)

	//开协程去消费
	go callback(ctx, msgChan)

	consumer, _ := os.Hostname()
	for {
		select {
		case <-ctx.Done():
			return nil
		case <-delayTicker.C:
			r.processDelayedMessages(ctx, topic)
		default:
			streams, err := r.client.XReadGroup(ctx, &redis.XReadGroupArgs{
				Group:    opt.ConsumerGroup,
				Consumer: fmt.Sprintf("consumer-%s", consumer),
				Streams:  []string{topic, ">"},
				Count:    opt.BatchNumber,
				Block:    1 * time.Second,
			}).Result()

			if err != nil {
				continue
			}
			for _, stream := range streams {
				for _, message := range stream.Messages {
					data, ok := message.Values["data"].(string)
					if !ok {
						continue
					}
					var msg queue.QueueMessgae
					if e := json.Unmarshal([]byte(data), &msg); e != nil {
						continue
					}
					//给消息添加一个id
					msg.Id = message.ID
					msgChan <- msg
				}
			}
		}
	}
}

func (r *RedisStreamDriver) Ack(ctx context.Context, q queue.IQueue, qm queue.QueueMessgae) {
	r.client.XAck(ctx, q.Topic(), q.Group(), qm.GetId())
}

func (r *RedisStreamDriver) Nack(ctx context.Context, q queue.IQueue, qm queue.QueueMessgae) {
	delivery := qm.GetHeader(queue.HeaderDelivery).(int)
	qm.SetHeader(queue.HeaderDelivery, delivery+1)
	//重新放入延时队列
	if message, err := json.Marshal(qm); err == nil {
		if err = r.Publish(ctx, q.Topic(), message, 5); err == nil {
			r.Ack(ctx, q, qm)
		}
	}
}

func (r *RedisStreamDriver) Reject(ctx context.Context, q queue.IQueue, qm queue.QueueMessgae) {
	if q.Dlq() {
		dlqName := r.deadLetterQueueName(q.Topic())
		messageBytes, _ := json.Marshal(qm)
		r.client.RPush(ctx, dlqName, messageBytes)
	}
	r.Ack(ctx, q, qm)
}

// queueName 获取队列名称
func (r *RedisStreamDriver) queueName(topic string) string {
	return fmt.Sprintf("queue:%s", topic)
}

// delayQueueName 获取延迟队列名称
func (r *RedisStreamDriver) delayQueueName(topic string) string {
	return fmt.Sprintf("queue:%s:delayed", topic)
}

func (r *RedisStreamDriver) deadLetterQueueName(topic string) string {
	return fmt.Sprintf("queue:%s:dlq", topic)
}
